package top.shaozuo.simple.reactive.streams;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public class DefaultSubscriber<T> implements Subscriber<T> {

	private Subscription subscription;

	private int requestCount = 0;

	private int maxRequest = 0;

	/**
	 * 可处理的请求数量
	 * 
	 * @param maxRequest
	 *            必须大于0
	 */
	public DefaultSubscriber(int maxRequest) {
		if (maxRequest < 1) {
			throw new IllegalStateException("maxRequest 必须大于0");
		}
		this.maxRequest = maxRequest;
	}

	@Override
	public void onSubscribe(Subscription s) {
		this.subscription = s;
	}

	@Override
	public void onNext(Object o) {
		//System.out.println(count);
		System.out.println("收到数据：" + o);
		requestCount++;
		if (requestCount >= maxRequest) { // 当到达数据阈值时，取消 Publisher 给当前 Subscriber 发送数据
			subscription.cancel();
			return;
		}
	}

	@Override
	public void onError(Throwable t) {
		System.out.println("遇到异常：" + t);
	}

	@Override
	public void onComplete() {
		System.out.println("收到数据完成");
	}
}
